package com.xrui.hbase.conversion;

import com.google.common.collect.ImmutableList;
import com.xrui.hbase.Constants;
import com.xrui.hbase.HBaseCell;
import com.xrui.hbase.model.RecordFields;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericRecord;

import java.util.List;

/**
 * Converts an HBase cell encoded as a Avro record.
 */
public class AvroCellConverter implements CellConverter {
    public static final String PROVIDER_NAME = "AvroConverter";
    /**
     * Returns the field list of avro schema
     * @param schema the schema of avro record
     * @return the list of avro fields
     */
    private static List<String> getAvroFields(Schema schema) {
        ImmutableList.Builder<String> builder = ImmutableList.builder();
        for (Schema.Field field : schema.getFields()) {
            builder.add(field.name());
        }
        return builder.build();
    }

    /** {@inheritDoc} */
    @Override
    public RecordFields convert(HBaseCell cell) {
        GenericRecord genericRecord = (GenericRecord) cell.getData();
        Schema schema = genericRecord.getSchema();
        List<String> fields = getAvroFields(schema);
        RecordFields.Builder builder = RecordFields.Builder.newBuilder();
        builder.putField(Constants.ROW_KEY_FIELD_NAME, cell.getEntityId().toString());
        builder.putField(Constants.COLUMN_NAME_FIELD_NAME, cell.getColumn().toString());
        builder.putField(Constants.TIMESTAMP_FIELD_NAME, cell.getTimestamp());
        for (String field : fields) {
            Object value = genericRecord.get(field);
            builder.putField(field, value);
        }

        return builder.build();
    }

    /** {@inheritDoc} */
    @Override
    public String getName() {
        return PROVIDER_NAME;
    }
}
